package com.hudsonmq.spring.impl;


import com.hudsonmq.spring.HudsonMQClientController;
import com.hudsonmq.spring.domain.consume.PushRequest;
import com.hudsonmq.spring.utils.ThreadPoolHelper;
import jakarta.annotation.PreDestroy;
import jakarta.annotation.Resource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Component;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

@Component
@ConditionalOnProperty(prefix = "hudson",name = "consume-group")
public class PushMessageServiceOneThread {

    private static final Logger logger = LoggerFactory.getLogger(PushMessageServiceOneThread.class);

    private static final BlockingQueue<PushRequest> REQUEST_QUEUE = new LinkedBlockingQueue<>();

    // 监控队列的最后一次消费时间,如果超过固定消费时间,则重新拉取消息
    private static final ConcurrentHashMap<String, PushRequest> PUSH_MESSAGE_WAIT_TIME = new ConcurrentHashMap<>();

    @Resource
    @Lazy
    private HudsonMQClientController consumeController;

    private static volatile boolean running = true;

    @PreDestroy
    public void onDestroy() {
        shutdown();
    }

    public void start() {
        ThreadPoolHelper.DEFAULT_THREAD_POOL.execute(() -> {
            while (running) {
                try {
                    PushRequest pushRequest = REQUEST_QUEUE.take();
                    if (!HudsonMQClientController.containsTopic(pushRequest)) {
                        PUSH_MESSAGE_WAIT_TIME.remove(pushRequest.getRemark());
                        return;
                    }
                    consumeController.getPushMessage(pushRequest.getChannel(), pushRequest).addListener(future -> {
                        if (future.isSuccess()) {
                            logger.debug("發送獲取請求成功");
                            pushRequest.setLastConsumeTime(System.currentTimeMillis());
                            PUSH_MESSAGE_WAIT_TIME.put(pushRequest.getRemark(), pushRequest);
                        } else {
                            logger.debug("發送獲取請求失敗");
                        }
                    });
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    running = false;
                }
            }
        });

        ThreadPoolHelper.DEFAULT_SCHEDULED_POOL.scheduleAtFixedRate(() -> {
            long l = System.currentTimeMillis();
            PUSH_MESSAGE_WAIT_TIME.forEach((k, v) -> {
                if (l - v.getLastConsumeTime() > 30_000 * 1.5) {
                    pushRequest(new PushRequest(v.getChannel(), v.getTopic(), v.getQueueId(), 1));
                }
            });
        }, 0, 5, TimeUnit.SECONDS);
    }

    public void shutdown() {
        running = false;
    }

    public static void pushRequest(PushRequest pushRequest) {
        REQUEST_QUEUE.add(pushRequest);
    }
}
